package com.iwhalecloud.uncc.mq.producer;

import com.iwhalecloud.uncc.dao.SysConfigDao;
import com.iwhalecloud.uncc.model.MqClusterConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class MqProducerManager {

    private static final String CTGMQ = "CTG";
    private static final String ROCKETMQ = "ROCKET";

    @Value("${mq.type}")
    private String mqType;
    @Value("${mq.cluster-code}")
    private String mqCode;
    @Autowired
    private SysConfigDao sysConfigDao;

    private static String _mqType;
    private static String _mqCode;
    private static SysConfigDao _sysConfigDao;

    /**
     * key为主题
     */
    private static Map<String, MqClusterConfig> topicConfigMap = new ConcurrentHashMap<>();
    /**
     * key为mq集群id
     */
    private static Map<String, MqProducer> configProducerMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init(){
        _mqType = mqType;
        _mqCode = mqCode;
        _sysConfigDao = sysConfigDao;
    }

    public static synchronized MqProducer createMqProducer(String topic, MqClusterConfig mqConfig) {
        try {
            MqProducer producer = configProducerMap.get(mqConfig.getMqCode());
            if(producer == null) {
                if (_mqType.equalsIgnoreCase(CTGMQ)) {
                    producer = createCtgMqProducer(mqConfig);
                } else if (_mqType.equalsIgnoreCase(ROCKETMQ)) {
                    producer = createRocketMqProducer(mqConfig);
                }
                configProducerMap.put(mqConfig.getMqCode(), producer);
            }
            topicConfigMap.put(topic, mqConfig);
            return producer;

        } catch (Exception e) {
            log.error("MqProducer 创建异常," + ExceptionUtils.getStackTrace(e));
        }
        return null;

    }

    private static CtgMqProducer createCtgMqProducer(MqClusterConfig mqConfig) {
        CtgMqProducer producer = new CtgMqProducer();
        producer.init(mqConfig);
        return producer;
    }

    private static RocketMqProducer createRocketMqProducer(MqClusterConfig mqConfig) {
        RocketMqProducer producer = new RocketMqProducer();
        producer.init(mqConfig);
        return producer;
    }


    public static MqProducer getProducer(String topic) {
        MqClusterConfig mqConfig = topicConfigMap.get(topic);
        if (mqConfig == null) {
            log.info("No Config of this topic:" + topic + " is initialized! Use default mq config...");
            mqConfig = _sysConfigDao.getMqClusterConfig(_mqCode);
            topicConfigMap.put(topic, mqConfig);
        }
        MqProducer producer = configProducerMap.get(mqConfig.getMqCode());
        if (producer == null) {
            log.info("No Producer of mqConfig [" + mqConfig.getMqCode() + "] is initialized!");
            return createMqProducer(topic, mqConfig);
        }
        return producer;
    }
}
